package cn.doitedu.day02

import org.apache.spark.rdd.RDD
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext, TaskContext}

//将mapSideCombine改成false
object T13_CombineByKeyDemo3 {

  def main(args: Array[String]): Unit = {

    //1.创建SparkConf
    val conf = new SparkConf().setAppName("MapPartitionsWithIndexDemo")
      .setMaster("local[4]")

    //2.创建SparkContext
    val sc = new SparkContext(conf)

    val lst = List(
      ("spark", 1), ("hadoop", 1), ("hive", 1), ("spark", 1),
      ("spark", 1), ("flink", 1), ("hbase", 1), ("spark", 1),
      ("kafka", 1), ("kafka", 1), ("kafka", 1), ("kafka", 1),
      ("hadoop", 1), ("flink", 1), ("hive", 1), ("flink", 1)
    )
    //通过并行化的方式创建RDD，分区数量为4
    val wordAndOne: RDD[(String, Int)] = sc.parallelize(lst, 4)

    //在上游，key第一次出现对value的处理逻辑
    val f1 = (x: Int) => {
      val stageId = TaskContext.get().stageId()
      val partitionId = TaskContext.getPartitionId()
      println(s"f1 在stage：$stageId, 分区：$partitionId")
      x
    }
    //在上游，相同的ke再次出现对value的处理逻辑
    val f2 = (a: Int, b: Int) => {
      val stageId = TaskContext.get().stageId()
      val partitionId = TaskContext.getPartitionId()
      println(s"f2 在stage：$stageId, 分区：$partitionId")
      a + b
    }
    //在下游，相同key的value的聚合逻辑
    val f3 = (m: Int, n: Int) => {
      val stageId = TaskContext.get().stageId()
      val partitionId = TaskContext.getPartitionId()
      println(s"f3 在stage：$stageId, 分区：$partitionId")
      m + n
    }

    val reduced: RDD[(String, Int)] = wordAndOne.combineByKey(
      f1, f2, f3,
      new HashPartitioner(wordAndOne.partitions.length),
      false
    )

    reduced.saveAsTextFile("out/out15")
  }

}